package main.test.GraphFramesAPI

import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import org.graphframes.GraphFrame

object subGraph {
  def main(args: Array[String]): Unit = {
    //  def constructGraph: GraphFrame = {
    //    The following example demonstrates how
    //    to create a GraphFrame from vertex and edge DataFrame
    val sparkConf = new SparkConf()
    sparkConf.setAppName("createGraph").setMaster("local[*]")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new SQLContext(sc)
    val v = sqlContext.createDataFrame(List(
      ("a", "Alice", 34),
      ("b", "Bob", 36),
      ("c", "Charlie", 30),
      ("d", "David", 29),
      ("e", "Esther", 32),
      ("f", "Fanny", 36),
      ("g", "Gabby", 60)
    )).toDF("id", "name", "age")

    // Edge DataFrame
    val e = sqlContext.createDataFrame(List(
      ("a", "b", "friend"),
      ("b", "c", "follow"),
      ("c", "b", "follow"),
      ("f", "c", "follow"),
      ("e", "f", "follow"),
      ("e", "d", "friend"),
      ("d", "a", "friend"),
      ("a", "e", "friend")
    )).toDF("src", "dst", "relationship")


    // Create a GraphFrame
    val g = GraphFrame(v, e)

    /**
     * In GraphX, the subgraph() method takes an edge triplet (edge,src vertex, and dst vertex, plus attributes)
     * and allows the user to select a subgraph based on triplet and vertex filters.
     *
     *
     * GraphFrames provide an even more powerful way to select
     * subgraphs based on a combination of motif finding and DataFrame filters.
     * It provides three helper methods for subgraph selection-- filterVertices(condition),
     * filterEdges(condition) and  dropIsolateVertices().
     *
     * Simple subgraph: vertex and edge filters:  How to select a sub graph based upon vertex and edge filters.
     *
     *
     */
    // Select subgraph of users older than 30, and relationships of type "friend".
    // Drop isolated vertices (users) which are not contained in any edges (relationships).
    val g1 = g.filterVertices("age > 30")
      .filterEdges("relationship = 'friend'").dropIsolatedVertices()
    g1.vertices.show()

    /**
     * Complex subgraph: How to select a subgraph based upon triplet filters which
     * operate on an edge and its src and dst vertices. This example could be extended to go
     * beyond triplets by using more complex motifs.
     */

    // Select subgraph based on edges "e" of type "follow"
    // pointing from a younger user "a" to an older user "b".
    val paths = {
      g.find("(a)-[e]->(b)")
        .filter("e.relationship = 'follow'")
        .filter("a.age < b.age")
    }

    paths.show()

    // "paths" contains vertex info. Extract the edges.
    val e2 = paths.select("e.src", "e.dst", "e.relationship")
    // In Spark 1.5+, the user may simplify this call:
    //  val e2 = paths.select("e.*")

    // Construct the subgraph
    val g2 = GraphFrame(g.vertices, e2)
    g2.vertices.show()
    g2.edges.show()


  }

}
